package org.carpediem.conman.server;

import org.carpediem.conman.Util;
import org.carpediem.conman.dao.ContactDao;
import org.carpediem.conman.entity.Contact;
import org.carpediem.conman.entity.Document;
import org.carpediem.conman.entity.Note;
import org.carpediem.conman.trash.S3;
import org.jets3t.service.S3Service;
import org.jets3t.service.S3ServiceException;
import org.jets3t.service.model.S3Bucket;
import org.jets3t.service.model.S3Object;

import java.io.ByteArrayInputStream;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * @author Mark Petrovic
 */
public class S3Persister extends Thread {
    private final Logger logger = Logger.getLogger(S3Persister.class.getName());

    private final String AWS_HOST = "s3.amazonaws.com";
    private S3Bucket s3DocumentBucket;
    private S3Bucket s3NoteBucket;
    private BlockingQueue<Contact> s3Queue = new LinkedBlockingQueue<Contact>();
    private volatile boolean spin = true;

    // bean properties
    private String documentBucket;
    private String noteBucket;
    private ContactDao contactDao;
    private S3Service s3Service;

    public S3Persister() {
        this.setDaemon(true);
        this.setName("S3Persister-Thread");
    }

    public void init() {
        try {
            s3DocumentBucket = s3Service.getOrCreateBucket(documentBucket);
            s3NoteBucket = s3Service.getOrCreateBucket(noteBucket);
            logger.info("Document and note buckets created");
        } catch (S3ServiceException e) {
            logger.log(Level.WARNING, "Cannot get/create bucket", e);
        }
        this.start();
    }

    public void run() {
        logger.info("S3Persister startup: " + this);

        while (spin) {
            try {
                Contact contact = s3Queue.take();
                try {
                    for (Document document : contact.getDocument()) {
                        if (!document.isPersistedToS3() && document.getBuffer() != null) {
                            logger.info("Saving contact/document to S3: " + document.getOriginalName());
                            S3Object output = s3Service.putObject(s3DocumentBucket, s3Object(document));
                            document.setPersistedToS3(true);
                            logger.info("Saved contact/document to S3: " + document.getOriginalName() + " " + output);
                        }
                    }
                    for (Note note : contact.getNote()) {
                        if (!note.isPersistedToS3()) {
                            logger.info("Saving note to S3: " + note.getNote());
                            S3Object output = s3Service.putObject(s3NoteBucket, s3Object(note));
                            note.setPersistedToS3(true);
                            logger.info("Saved note to S3: " + note.getNote() + ": " + output);
                        }
                    }
                    contactDao.update(contact);
                } catch (S3ServiceException e) {
                    s3Queue.put(contact);
                    logger.log(Level.WARNING, "S3 exception persisting contact content.  Placing back on queue.", e);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException("Unexpected interruption while processing queue");
            }
        }
    }

    private S3Object s3Object(Document document) {
        byte[] digest = Util.digest(document.getBuffer());
        String s3FileName = Util.byteToHex(digest);

        S3Object s3Object = new S3Object(s3FileName);
        s3Object.setBucketName(S3.documentBucketName);
        s3Object.setDataInputStream(new ByteArrayInputStream(document.getBuffer()));
        s3Object.setContentLength(document.getBuffer().length);
        s3Object.setContentType(document.getMimeType());
        s3Object.setMd5Hash(digest);
        s3Object.addMetadata(Contact.CONTACT, document.getContact().getEmailAddress());
        s3Object.addMetadata(Document.ORIGINAL_NAME, document.getOriginalName());

        return s3Object;
    }

    private S3Object s3Object(Note note) {
        byte[] buffer = Util.bytes(note.getNote());

        S3Object s3Object = new S3Object(note.getS3Key());
        s3Object.setBucketName(noteBucket);
        s3Object.setDataInputStream(new ByteArrayInputStream(buffer));
        s3Object.setContentLength(note.getNote().length());
        s3Object.setContentType("text/plain");
        s3Object.setMd5Hash(Util.digest(buffer));
        s3Object.addMetadata(Contact.CONTACT, note.getContact().getEmailAddress());

        return s3Object;
    }

    public void enqueue(Contact contact) {
        try {
            s3Queue.put(contact);
            logger.info("Queued contact: " + contact);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("Unexpected interruption while put'ing contact on queue");
        }
    }

    public URI documentUri(byte[] buffer) {
        URI uri = null;
        try {
            uri = new URI("https://" + documentBucket + "." + AWS_HOST + "/" + Util.hexDigest(buffer));
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }
        return uri;
    }

    public URI noteUri(Note note) {
        URI uri = null;
        try {
            uri = new URI("https://" + noteBucket + "." + AWS_HOST + "/" + note.getS3Key());
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }
        return uri;
    }

    public void quit() {
        spin = false;
    }

    public ContactDao getContactDao() {
        return contactDao;
    }

    public void setContactDao(ContactDao contactDao) {
        this.contactDao = contactDao;
    }

    public S3Service getS3Service() {
        return s3Service;
    }

    public void setS3Service(S3Service s3Service) {
        this.s3Service = s3Service;
    }

    public String getDocumentBucket() {
        return documentBucket;
    }

    public void setDocumentBucket(String documentBucket) {
        this.documentBucket = documentBucket;
    }

    public String getNoteBucket() {
        return noteBucket;
    }

    public void setNoteBucket(String noteBucket) {
        this.noteBucket = noteBucket;
    }

}
